package com.alibaba.nacos.naming.core.v2.client;

import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.naming.core.v2.event.client.ClientEvent.ClientChangeEvent;
import com.alibaba.nacos.naming.core.v2.pojo.BatchInstancePublishInfo;
import com.alibaba.nacos.naming.core.v2.pojo.InstancePublishInfo;
import com.alibaba.nacos.naming.core.v2.pojo.Service;
import com.alibaba.nacos.naming.pojo.Subscriber;
import com.alibaba.nacos.naming.utils.DistroUtils;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public abstract class AbstractClient
        implements Client {

    protected final ConcurrentHashMap<Service, InstancePublishInfo> publishers = new ConcurrentHashMap<>(16, 0.75f, 1);
    private final ConcurrentHashMap<Service, Subscriber> subscribers = new ConcurrentHashMap<>(16, 0.75f, 1);
    @Getter
    protected volatile long lastUpdatedTime = System.currentTimeMillis();;
    @Setter
    @Getter
    protected final AtomicLong revision;
    @Setter
    @Getter
    protected ClientAttributes attributes;

    public AbstractClient(Long revision) {
        this.revision = new AtomicLong(Objects.requireNonNullElse(revision, 0L));
    }

    @Override
    public void setLastUpdatedTime() {
        this.lastUpdatedTime = System.currentTimeMillis();
    }

    @Override
    public long recalculateRevision() {
        int hash = DistroUtils.hash(this);
        revision.set(hash);
        return hash;
    }

    @Override
    public boolean addServiceInstance(Service service,
                                      InstancePublishInfo instancePublishInfo) {

        if (instancePublishInfo instanceof BatchInstancePublishInfo) {
            throw new UnsupportedOperationException();
        } else {
            publishers.put(service, instancePublishInfo);
            log.info("ClientId:{} add publisher:{}", getClientId(), instancePublishInfo);
        }

        NotifyCenter.publishEvent(new ClientChangeEvent(this));
        return true;
    }

    @Override
    public InstancePublishInfo removeServiceInstance(Service service) {
        InstancePublishInfo removed = publishers.remove(service);
        log.info("ClientId:{} remove publisher:{}", getClientId(), removed);
        Optional.ofNullable(removed).ifPresent(instancePublishInfo -> NotifyCenter.publishEvent(new ClientChangeEvent(this)));
        return removed;
    }

    @Override
    public InstancePublishInfo getInstancePublishInfo(Service service) {
        return publishers.get(service);
    }

    @Override
    public Collection<Service> getAllPublishedService() {
        return publishers.keySet();
    }

    @Override
    public Subscriber getSubscriber(Service service) {
        return subscribers.get(service);
    }

    @Override
    public boolean addServiceSubscriber(Service usingService,
                                        Subscriber subscriber) {
        subscribers.put(usingService, subscriber);
        return true;
    }

    @Override
    public boolean removeServiceSubscriber(Service usingService) {
        subscribers.remove(usingService);
        return true;
    }

    @Override
    public ClientSyncData generateSyncData() {
        List<String> namespaces = new LinkedList<>();
        List<String> groupNames = new LinkedList<>();
        List<String> serviceNames = new LinkedList<>();
        List<InstancePublishInfo> instances = new LinkedList<>();

        for (Map.Entry<Service, InstancePublishInfo> entry : publishers.entrySet()) {
            InstancePublishInfo value = entry.getValue();
            if (value instanceof BatchInstancePublishInfo) {
                throw new UnsupportedOperationException();
            }

            namespaces.add(entry.getKey().getNamespace());
            groupNames.add(entry.getKey().getGroup());
            serviceNames.add(entry.getKey().getName());
            instances.add(value);
        }

        ClientSyncData clientSyncData = new ClientSyncData().setClientId(getClientId())
                .setNamespaces(namespaces)
                .setGroupNames(groupNames)
                .setServiceNames(serviceNames)
                .setInstancePublishInfos(instances);
        clientSyncData.getAttributes().addClientAttribute("revision", getRevision());

        return clientSyncData;
    }

    @Override
    public void setRevision(long revision) {
        this.revision.set(revision);
    }
}
